// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_MULTI_ROW_MUTATOR_HPP_
#define _XBASE_MULTI_ROW_MUTATOR_HPP_

#include <vector>

#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/intrusive_ptr.hpp>
#include <boost/system/error_code.hpp>

#include "protocol.hpp"
#include "location_cache.hpp"
#include "session_manager.hpp"
#include "row_mutation_data.hpp"
#include "multi_row_mutator_on_tablet.hpp"

namespace xbase
{
	class multi_row_mutator
		: public async_module_base<io_service_pool::user_level, io_service_pool::xbase_level>
	{
	public:
		typedef async_module_base<io_service_pool::user_level, io_service_pool::xbase_level> base_type;
		typedef multi_row_mutator this_type;
		typedef boost::intrusive_ptr<this_type> ptr_type;

		typedef session_manager::session_ptr session_ptr;
		typedef multi_row_mutator_on_tablet::ptr_type multi_row_mutator_on_tablet_ptr;

		typedef boost::system::error_code error_code;

		typedef boost::function<void(std::size_t, const error_code&)> handler_type;

		typedef std::vector<row_mutation_data_ptr> container_type;
		typedef boost::shared_ptr<container_type> container_ptr;
		typedef container_type::iterator container_iterator;

		typedef protocol::packet_type<protocol::BATCH_WRITE_REQ>::type batch_write_req;
		typedef protocol::packet_type<protocol::BATCH_WRITE_REQ>::ptr_type batch_write_req_ptr;

		typedef protocol::packet_type<protocol::BATCH_WRITE_RESP>::type batch_write_resp;
		typedef protocol::packet_type<protocol::BATCH_WRITE_RESP>::ptr_type batch_write_resp_ptr;

		struct comp
		{
			bool operator()(const row_mutation_data_ptr &ptr,
					const std::string &row_name) const {
				return ptr->row_name() < row_name;
			}
			bool operator()(const std::string &row_name,
					const row_mutation_data_ptr &ptr) const {
				return row_name < ptr->row_name();
			}
		};

	public:
		multi_row_mutator(io_service_pool &pool,
				  session_manager &manager,
				  location_cache &cache)
			: base_type(pool, "multi_row_mutator")
			, m_io_service_pool(pool)
			, m_options(pool.options())
			, m_session_manager(manager)
			, m_location_cache(cache)
			, m_success_row(0)
			, m_tablet_info_invalid(false) {}

			// 将修改操作应用到指定表
			//
			// 用户未指定的时间戳将在此时取当前系统时间来代替
			//
		void async_apply(const std::string &table_name,
				 const container_ptr &mutations,
				 uint64_t timestamp,
				 handler_type handler);

		void async_apply(const std::string &table_name,
				 const container_ptr &mutations,
				 handler_type handler);
	protected:
		void on_async_apply();
		
		void on_got_tablet(location_cache_value meta, const error_code &ec);

		void on_session_connected(session_ptr session, const error_code &ec);

		void on_complete(std::size_t success_row, const error_code &ec);

		void on_retry(std::size_t success_row, const error_code &ec);

		void on_error(std::size_t success_row, const error_code &ec);

		void dispatch(std::size_t success_row, const error_code &ec);

		ptr_type this_ptr() {
			return ptr_type(this);
		}

	private:
		io_service_pool &m_io_service_pool;
		const client_options &m_options;
		session_manager &m_session_manager;
		location_cache &m_location_cache;

		std::string m_table_name;
		container_ptr m_mutations;
		uint64_t m_timestamp;
		handler_type m_handler;
		std::size_t m_success_row;

		container_iterator m_start_pos;
		container_iterator m_end_pos;
		batch_write_req_ptr m_req_ptr;

		boost::shared_ptr<boost::asio::deadline_timer> m_timer;
		std::size_t m_retry_times;

		bool m_tablet_info_invalid;
		multi_row_mutator_on_tablet_ptr m_tablet_mutator;

		location_cache_value m_meta_data;
	};

	// template<typename UpperLevel,
	// 	 typename LowerLevel,
	// 	 typename LocationCache,
	// 	 typename MultiRowMutatorOnTablet>
	// class basic_multi_row_mutator
	// 	: public async_module_base<UpperLevel, LowerLevel>
	// {
	// public:
	// 	typedef async_module_base<UpperLevel, LowerLevel> base_type;
	// 	typedef basic_multi_row_mutator_helper<UpperLevel, LowerLevel, LocationCache, MultiRowMutatorOnTablet> this_type;
	// 	typedef boost::intrusive_ptr<this_type> ptr_type;

	// 	typedef LocationCache location_cache_type;
	// 	typedef typename location_cache_type::session_manager_type session_manager_type;
	// 	typedef typename session_manager_type::session_ptr session_ptr;
	// 	typedef MultiRowMutatorOnTablet multi_row_mutator_on_tablet_type;
	// 	typedef typename multi_row_mutator_on_tablet_type::ptr_type multi_row_mutator_on_tablet_ptr;

	// 	typedef boost::system::error_code error_code;

	// 	typedef boost::function<void(std::size_t, const error_code&)> handler_type;

	// 	typedef std::map<std::string, row_mutation_data_ptr> row_map;
	// 	typedef boost::shared_ptr<row_map> row_map_ptr;

	// 	typedef basic_multi_row_mutator_helper<UpperLevel, LowerLevel,
	// 					       location_cache_type,
	// 					       multi_row_mutator_on_tablet_type> helper;

	// public:
	// 	template<typename Handler>
	// 	void async_apply(const std::string table_name,
	// 			 const row_map_ptr &mutations,
	// 			 uint64_t timestamp,
	// 			 Handler handler) {
	// 		if(mutations->empty())
	// 		{
	// 			dispatch(0, error_code());
	// 			return;
	// 		}

	// 		m_table_name = table_name;
	// 		m_mutations = mutations;
	// 		m_timestamp = timestamp;
	// 		m_handler = handler;

	// 		on_apply();
	// 	}

	// protected:
	// 	void on_apply() {
	// 		if(m_mutations->empty())
	// 		{
	// 			on_apply_alive = false;
	// 		}
	// 		else
	// 		{
	// 			on_apply_alive = true;
	// 			const std::string &first_row = m_mutations->begin()->first;
	// 			m_location_cache.async_get_meta(boost::bind(&this_type::on_got_tablet_info,
	// 								    this_ptr(),
	// 								    _1, _2, _3),
	// 							m_table_name,
	// 							first_row,
	// 							false,
	// 							0);
	// 		}
	// 	}

	// 	void on_retry_got_tablet_info() {
	// 		if(! m_timer_got_tablet_info)
	// 			m_timer_got_tablet_info->reset(new boost::asio::deadline_timer(this->get_lower_level_service()));
	// 		m_timer_got_tablet_info->expires_from_now(m_options.get_retry_interval());
	// 		m_timer_got_tablet_info->async_wait(boost::bind(&this_type::on_apply, this_ptr()));
	// 	}

	// 	void on_got_tablet_info(protocol::tablet_info_ptr tablet_info,
	// 				protocol::server_address_ptr address,
	// 				const error_code &ec) {
	// 		if(ec)
	// 		{
	// 			if((ec == error::tablet_not_exists) ||
	// 			   (ec == error::table_not_exists))
	// 			{
	// 				on_retry_got_tablet_info();
	// 			}
	// 			else
	// 			{
	// 				on_error(0, ec);
	// 			}
	// 			return;
	// 		}

	// 		const row_map::iterator start = m_mutations->lower_bound(tablet_info->m_start_row);
	// 		row_map::iterator end;
	// 		if(tablet_info->m_end_row.empty())
	// 		{
	// 			end = m_mutations.end();
	// 		}
	// 		else
	// 		{
	// 			end = m_row_map.lower_bound(tablet_info->m_end_row);
	// 		}

	// 		row_vector_ptr rows(new row_vector);
	// 		for(row_map::iterator it = start; it != end; ++it)
	// 		{
	// 			rows->push_back(it->second);
	// 		}
	// 		m_row_map.erase(start, end);

	// 		helper_ptr ptr;
	// 		if(m_free_helper.empty())
	// 		{
	// 			ptr.reset(new_async_module(helper, m_pool, m_session_manager, m_location_cache));
	// 		}
	// 		else
	// 		{
	// 			ptr = m_free_helper.back();
	// 			m_free_helper.pop_back();
	// 		}

	// 		commit_item item;
	// 		item.m_tablet_info = tablet_info;
	// 		item.m_address = address;
	// 		item.m_rows = rows;
	// 		item.m_ptr = ptr;

	// 		typedef std::map<uint64_t, commit_item>::value_type val_type;
	// 		typedef std::map<uint64_t, commit_item>::iterator iter_type;

	// 		std::pair<iter_type, bool> ret =
	// 			m_commit_map.insert(val_type(tablet_info->m_tablet_id,
	// 						     item));
	// 		assert(ret.second); void(ret);

	// 		m_session_manager.async_connect(address->to_endpoint(),
	// 						boost::bind(&this_type::on_connected,
	// 							    this_ptr(),
	// 							    tablet_info->m_tablet_id,
	// 							    rows,
	// 							    _1, _2));
	// 	}

	// 	void on_connected(uint64_t id, row_map_ptr rows,
	// 			  session_ptr session,
	// 			  const error_code &ec) {
	// 		if(ec)
	// 		{
	// 			recover_rows(rows, 0);
	// 			on_retry_got_tablet_info();
	// 		}
	// 		else
	// 		{
	// 			ptr->async_apply(boost::bind(&this_type::on_complete,
	// 						     this_ptr(), rows,
	// 						     _1, _2));
	// 			on_apply();
	// 		}
	// 	}

	// 	void on_complete(row_vector_ptr rows,
	// 			 std::size_t success_rows, const error_code &ec) {
	// 		m_success_row += success_rows;
	// 		if(ec)
	// 		{
	// 			if(success_rows < rows->size())
	// 			{
	// 				for(std::size_t i = success_rows; i < rows->size(); ++i)
	// 				{
	// 					const row_mutation_data_ptr &ptr_to_add = ((*rows)[i]);
	// 					const std::pair<row_map::iterator, bool> ret =
	// 						m_mutations->insert(row_map::value_type(ptr_to_add->row_name(),
	// 											ptr_to_add));
	// 					assert(ret.second); void(ret);
	// 				}
	// 				if(! on_apply_alive)
	// 				{
	// 					on_apply();
	// 				}
	// 			}
	// 		}
	// 		else if(success_rows != rows->size())
	// 		{
	// 			LOG(ERROR, "");
	// 		}
	// 		else
	// 		{

	// 		}
	// 	}

	// private:
	// 	io_service_pool &m_io_service_pool;
	// 	const client_options &m_options;
	// 	session_manager_type &m_session_manager;
	// 	location_cache_type &m_location_cache;

	// 	std::string m_table_name;
	// 	row_map_ptr m_mutations;
	// 	uint64_t m_timestamp;
	// 	handler_type m_handler;
	// 	std::size_t m_success_row;

	// 	struct commit_item
	// 	{
	// 		protocol::tablet_info_ptr m_tablet_info;
	// 		protocol::server_address_ptr m_address;
	// 		row_vector_ptr m_rows;
	// 		helper_ptr m_ptr;
	// 	};
	// 	std::map<uint64_t, commit_item> m_commit_map;
	// 	std::vector<helper_ptr> m_free_helper;

	// 	boost::shared_ptr<boost::asio::deadline_timer> m_timer_got_tablet_info;
	// 	std::size_t m_retry_times_got_tablet_info;

	// 	boost::shared_ptr<boost::asio::deadline_timer> m_timer;
	// 	std::size_t m_retry_times;

	// 	protocol::tablet_info_ptr m_tablet_info;
	// 	protocol::server_address_ptr m_server_address;
	// 	bool m_tablet_info_invalid;
	// 	uint64_t m_tablet_info_timestamp;
	// 	session_ptr m_session;
	// 	multi_row_mutator_on_tablet_ptr m_tablet_mutator;

	// };

	typedef multi_row_mutator::ptr_type multi_row_mutator_ptr;

	template<typename Handler>
	inline
	void async_apply(io_service_pool &pool,
			 session_manager &manager,
			 location_cache &cache,
			 const std::string &table_name,
			 const multi_row_mutator::container_ptr &mutations,
			 uint64_t timestamp,
			 Handler handler) {
		multi_row_mutator_ptr mutator(new_async_module(multi_row_mutator, pool, manager, cache));
		mutator->async_apply(table_name, mutations,
				     timestamp, handler);
	}

	template<typename Handler>
	inline
	void async_apply(io_service_pool &pool,
			 session_manager &manager,
			 location_cache &cache,
			 const std::string &table_name,
			 const multi_row_mutator::container_ptr &mutations,
			 Handler handler) {
		multi_row_mutator_ptr mutator(new_async_module(multi_row_mutator, pool, manager, cache));
		mutator->async_apply(table_name, mutations,
				     handler);
	}

} // namespace xbase

#endif	// _XBASE_MULTI_ROW_MUTATOR_HPP_
